package com.gitee.sparta.core.cluster.nio;

import cn.hutool.core.collection.CollUtil;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.bean.cluster.ServerBean;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.cluster.strategy.StrategyFactory;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;

/**
 * <p>
 * leader nio
 * </p>
 *
 * @author zyred
 * @since v 0.0.2
 * @deprecated 弃用 nio 作为通讯组件
 */
@Slf4j
@Deprecated
public class NioClusterLeader extends AbstractClusterLeader {

    private Selector selector;
    private ServerSocketChannel ssc;


    public NioClusterLeader(SpartaCoreProperties properties,
                            RedissonClient client,
                            LoadBalance loadBalance) {
        super(properties, client, loadBalance);
    }


    @Override
    public void buildLeader() {
        try {
            selector = Selector.open();
            ssc = ServerSocketChannel.open();
            ssc.bind(new InetSocketAddress(0));
            ssc.configureBlocking(false);
            InetSocketAddress addr = (InetSocketAddress) ssc.getLocalAddress();

            properties.getCluster().setClusterTransportPort(addr.getPort());

            ssc.register(selector, SelectionKey.OP_ACCEPT);
            this.buildProcessorThread();
        } catch (IOException ex) {
            log.error("[Sparta] Nio leader enable error : {}", ex.getMessage(), ex);
            return;
        }
        log.info("[Sparta] Nio leader enabled, ctp: {}", properties.getCluster().getClusterTransportPort());
    }


    @Override
    public void demotionFollower(String ip, int port) {

    }

    @SuppressWarnings("all")
    public void buildProcessorThread() {
        new Thread(() -> {
            while (!Thread.interrupted()) {
                try {
                    selector.select();
                    Iterator<SelectionKey> ite = selector.selectedKeys().iterator();
                    while (ite.hasNext()) {
                        SelectionKey key = ite.next();
                        ite.remove();

                        // 连接处理
                        if (key.isAcceptable()) {
                            this.processorAccept(ssc, selector);
                        }
                        // 读取处理
                        else if (key.isReadable()) {
                            SocketChannel client = (SocketChannel) key.channel();
                            try {
                                this.handlerReads(client, selector);
                            } catch (IOException ex) {
                                this.handlerReadError(client);
                            }
                        }
                    }
                } catch (Exception ex) {
                    log.error("[Sparta] Nio process communication error : {}", ex.getMessage(), ex);
                }
            }
        }, "nio-leader").start();
    }

    /**
     * 处理读取数据失败的错误
     *
     * @param client            客户端连接
     * @throws IOException      IOException
     */
    private void handlerReadError(SocketChannel client) throws IOException {
        // 剔除从节点
        Integer sid = ClusterCache.getSid(client);
        if (Objects.nonNull(sid)) {
            this.removeFollower(sid);
            this.loadBalance.removeNode(sid);
            ClusterCache.removeCli(client, sid);
            log.warn("[Sparta] Nio client disconnect : {}", client.getRemoteAddress());
        }
        client.close();
    }


    /**
     * 处理连接事件, 连接上了，就需要注册 follower
     *
     * @param ssc      server socket channel
     * @param selector selector
     */
    private void processorAccept(ServerSocketChannel ssc, Selector selector) throws IOException {
        SocketChannel cli = ssc.accept();
        cli.configureBlocking(false);
        cli.register(selector, SelectionKey.OP_READ);
    }


    /**
     * 剔除服务
     *
     * @param sid 服务编号
     */
    private void removeFollower(Integer sid) {
        RLock lock = this.client.getLock(this.properties.getLock());
        try {
            lock.lock();
            RMap<Integer, ServerBean> serverTable = this.client.getMap(this.properties.getServerTable());
            serverTable.remove(sid);
        } finally {
            lock.unlock();
        }
    }

    /**
     * 读取客户端发送给 leader 的数据
     *
     * @param client   客户端连接通道
     * @param selector leader 选择器
     * @throws IOException ioe
     */
    private void handlerReads(SocketChannel client, Selector selector) throws IOException {
        List<ClusterProtocol> protocolList = HandlerReadWriteHelper.readProtocol(client);
        if (CollUtil.isEmpty(protocolList)) {
            return;
        }

        for (ClusterProtocol protocol : protocolList) {
            StrategyFactory.getStrategy(protocol.getCmd()).handle(null, client, protocol);
        }
        client.register(selector, SelectionKey.OP_READ);
    }


    @Override
    public void doWriteFollower(ClusterProtocol protocol) throws Exception {
        HandlerReadWriteHelper.writeProtocol(protocol, protocol.getCli());
    }
}
